/**
 * Copyright 2013 Netflix, Inc.
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.atom.zx.operators;

import java.util.concurrent.atomic.AtomicReference;

import com.atom.zx.MainTest;
import com.atom.zx.Subscription;
import com.atom.zx.util.Logs;
import org.junit.Test;

public final class SafeObservableSubscription implements Subscription {

    private static final Subscription UNSUBSCRIBED = new Subscription() {
        @Override
        public void unsubscribe() {
        }
    };
    private final AtomicReference<Subscription> actualSubscription = new AtomicReference<Subscription>();

    public SafeObservableSubscription() {
    }

    public SafeObservableSubscription(Subscription actualSubscription) {
        this.actualSubscription.set(actualSubscription);
    }


    public SafeObservableSubscription wrap(Subscription actualSubscription) {
        // 判断该并发容器是否为null , 如果不为null ,那么直接就赋值了,
        if (!this.actualSubscription.compareAndSet(null, actualSubscription)) {
            // 实际上判断的actualSubscription == UNSUBSCRIBED
            if (this.actualSubscription.get() == UNSUBSCRIBED) {
                // 如果等于 就直接进行 执行,并返回
                actualSubscription.unsubscribe();
                return this;
            }
            // 如果该订阅不为 UNSUBSCRIBED 直接报错 不能多次设置订阅。
            throw new IllegalStateException("Can not set subscription more than once.");
        }
        return this;
    }

    @Override
    public void unsubscribe() {
        Subscription actual = actualSubscription.getAndSet(UNSUBSCRIBED);
        if (actual != null) {
            actual.unsubscribe();
        }
    }

    public boolean isUnsubscribed() {
        return actualSubscription.get() == UNSUBSCRIBED;
    }

    public static class UnitTest {

        public void testWrapAfterUnsubscribe() {
            SafeObservableSubscription atomicObservableSubscription = new SafeObservableSubscription();
            atomicObservableSubscription.unsubscribe();
            Subscription innerSubscription = () -> Logs.e("testWrapAfterUnsubscribe", "unsubscribe()");
            atomicObservableSubscription.wrap(innerSubscription);
            atomicObservableSubscription.unsubscribe();

            atomicObservableSubscription.wrap(innerSubscription);
            atomicObservableSubscription.wrap(innerSubscription);
            atomicObservableSubscription.wrap(innerSubscription);
            atomicObservableSubscription.wrap(innerSubscription);

        }
    }

    //141181199309210070
    public static final void main(String[] aa) {
        SafeObservableSubscription.UnitTest test = new SafeObservableSubscription.UnitTest();
        test.testWrapAfterUnsubscribe();
    }

}
